package com.rran.study.chat.friend.mq;

import com.alibaba.fastjson.JSON;
import com.rran.study.chat.friend.mq.entity.SyscUserInfoMqEntity;
import com.rran.study.chat.friend.service.UserService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;

/**
 * @author yy
 * @Type SyncUpdateUserInfo.java
 * @Desc 同步用户基础信息
 * @date 2020/8/19
 */
@Component
@Slf4j
public class SyncUpdateUserInfo {

    @Resource
    private UserService userService;

//    @KafkaListener(groupId = "${user.kafka.group.syncUserInfo}", topics = "${user.kafka.topic.syncUserInfo}")
    public void consume(ConsumerRecord<String, String> record, Acknowledgment ack) {
        log.info("===>>> 功能:【同步更新用户信息】, 消费kafka消息:{}", record.value());

        // 解析实体
        SyscUserInfoMqEntity userMQEntity = JSON.parseObject(record.value(), SyscUserInfoMqEntity.class);
        if (userMQEntity == null) {
            ack.acknowledge();
        }
        //同步用户信息入库
        userService.syncUpdateUserInfo(userMQEntity);
        ack.acknowledge();
    }

}
